package com.atguigu.edu.realtime.app.dwd.user;


import com.atguigu.edu.realtime.app.BaseAppSql;
import com.atguigu.edu.realtime.common.Constant;
import com.atguigu.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

//当日用户注册   用户域用户注册实时表
//数据来源 Topic ods_db    user_info insert
//数据去处 Topic dwd_user_register
public class DwdUserUserRegister extends BaseAppSql {
    public static void main(String[] args) {
        new DwdUserUserRegister().init(3702,2,"DwdUserUserRegister");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //1.读取Kafka ods_db
        sourceFromOdsDb(tEnv,"DwdUserUserRegister");

        //2.过滤出 user_info 新增用户数据
        Table result = tEnv.sqlQuery("select \n" +
                "    `data`['id'] user_id,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `ts`*1000 ts \n" +
                "from ods_db \n" +
                "where `database` = 'edu' \n" +
                "    and `table` = 'user_info' \n" +
                "    and `type` = 'insert'");
        //3.创建动态表和Kafka topic 关联
        tEnv.executeSql("create table dwd_user_register(\n" +
                "    user_id string,\n" +
                "    create_time string,\n" +
                "    ts bigint" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_USER_REGISTER));
        //4.往动态表插入写出数据
        result.executeInsert("dwd_user_register");
    }
}
